package com.zhang.gmall.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.sun.xml.internal.bind.v2.TODO;
import com.zhang.gmall.app.func.DimAsyncFunction;
import com.zhang.gmall.beans.OrderWide;
import com.zhang.gmall.beans.UserProvinceSku;
import com.zhang.gmall.utils.ClickHouseUtil;
import com.zhang.gmall.utils.DateTimeUtil;
import com.zhang.gmall.utils.KafkaUtil;
import com.zhang.gmall.utils.MyEnv;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;
import java.util.concurrent.TimeUnit;

/**
 * @title: 用户省份商品（sku）统计分析
 * @author: zhang
 * @date: 2022/3/27 20:07
 */
public class UserProvinceSku10sApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取环境
        StreamExecutionEnvironment env = MyEnv.getStreamingEnv(4, false);

        //TODO 2.读取kafka dwd_order_wider主题数据，转化为POJO类
        // 使用过滤脏数据，因为上油为upsert-kafka可能存在value为null的数据,直接丢弃
        String topic = "dwd_order_wide";
        String groupId = "user_province_sku_10s_app";
        SingleOutputStreamOperator<UserProvinceSku> userProvinceSkuNODim = env
                .addSource(KafkaUtil.getKafkaSource2(topic, groupId))
                .process(new ProcessFunction<String, OrderWide>() {
                    @Override
                    public void processElement(String value, ProcessFunction<String, OrderWide>.Context ctx, Collector<OrderWide> out) throws Exception {
                        try {
                            OrderWide orderWide = JSON.parseObject(value, OrderWide.class);
                            out.collect(orderWide);
                        } catch (Exception e) {
                            System.out.println("发现脏数据：" + value);
                        }
                    }
                })
                //TODO 3.过滤order_detail_id的第一条数据，因为上游是upsert-kafka，
                // 在leftjoin的时候第一条没有活动和优惠券信息，之后会发一个value为null的代表撤回信息，然后才是关联到活动或者是优惠券的数据
                // 这里不需要活动和优惠券数据所以要过滤同一个下order_detail_id的第一个数据
                .keyBy(OrderWide::getOrder_detail_id)
                .process(new KeyedProcessFunction<Long, OrderWide, UserProvinceSku>() {
                    //定义一个状态标记是一条数据是否存在
                    private ValueState<Boolean> exist;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //设置状态ttl，因为做join的时候状态保存5s，也就是说在5s内要么join上要么join不上，所以这里状态不需要保存过长时间，和上游一致
                        //同时当读写和创建的时候重置ttl时间
                        ValueStateDescriptor<Boolean> valueStateDescriptor = new ValueStateDescriptor<>("is-exist", Types.BOOLEAN);
                        StateTtlConfig ttlConfig = new StateTtlConfig.Builder(Time.seconds(5))
                                .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
                                .build();
                        valueStateDescriptor.enableTimeToLive(ttlConfig);
                        exist = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(OrderWide orderWide, KeyedProcessFunction<Long, OrderWide, UserProvinceSku>.Context ctx, Collector<UserProvinceSku> out) throws Exception {
                        if (exist.value() == null) {
                            exist.update(true);
                            //TODO 4.将数据转换为新的UserProvinceSku POJO对象
                            out.collect(UserProvinceSku.builder()
                                    .sku_id(orderWide.getSku_id())
                                    .sku_name(orderWide.getSku_name())
                                    .sku_price(orderWide.getOrder_price())
                                    .user_id(orderWide.getUser_id())
                                    .province_id(orderWide.getProvince_id())
                                    .order_sku_num(orderWide.getSku_num())
                                    .order_amount(orderWide.getTotal_amount())
                                    .ts(DateTimeUtil.toTs(orderWide.getCreate_time()))
                                    .order_ct(1L)
                                    .build());
                        }
                    }
                })
                //TODO 5.提取时间戳生成Watermark,分组开窗聚合
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<UserProvinceSku>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner(new SerializableTimestampAssigner<UserProvinceSku>() {
                                    @Override
                                    public long extractTimestamp(UserProvinceSku element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                })
                )
                .keyBy(new KeySelector<UserProvinceSku, Tuple3<Long, Long, Long>>() {
                    @Override
                    public Tuple3<Long, Long, Long> getKey(UserProvinceSku value) throws Exception {
                        return Tuple3.of(
                                value.getSku_id(),
                                value.getProvince_id(),
                                value.getUser_id()
                        );
                    }
                })
                .window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)))
                .reduce(new ReduceFunction<UserProvinceSku>() {
                            @Override
                            public UserProvinceSku reduce(UserProvinceSku value1, UserProvinceSku value2) throws Exception {
                                value1.setOrder_sku_num(value1.getOrder_sku_num() + value2.getOrder_sku_num());
                                value1.setOrder_amount(value1.getOrder_amount().add(value2.getOrder_amount()));
                                value1.setOrder_ct(value1.getOrder_ct() + value2.getOrder_ct());
                                return value1;
                            }
                        },
                        new WindowFunction<UserProvinceSku, UserProvinceSku, Tuple3<Long, Long, Long>, TimeWindow>() {
                            @Override
                            public void apply(Tuple3<Long, Long, Long> longLongLongTuple3, TimeWindow window, Iterable<UserProvinceSku> input, Collector<UserProvinceSku> out) throws Exception {
                                UserProvinceSku userProvinceSku = input.iterator().next();
                                //补充窗口时间信息
                                userProvinceSku.setStt(DateTimeUtil.toYMDhms(new Date(window.getStart())));
                                userProvinceSku.setEdt(DateTimeUtil.toYMDhms(new Date(window.getEnd())));
                                //添加处理时间戳
                                userProvinceSku.setTs(System.currentTimeMillis());
                                out.collect(userProvinceSku);
                            }
                        });
        //打印测试
        //userProvinceSkuNODim.print("noDim>>>>>>");

        //TODO 6.关联维度（异步IO）
        // 关联用户维度
        SingleOutputStreamOperator<UserProvinceSku> userProvinceSkuWithUser = AsyncDataStream.unorderedWait(userProvinceSkuNODim,
                new DimAsyncFunction<UserProvinceSku>("DIM_USER_INFO") {

                    @Override
                    public String getKey(UserProvinceSku input) {
                        return input.getUser_id().toString();
                    }

                    @Override
                    public void join(UserProvinceSku input, JSONObject dimInfo) throws ParseException {
                        //补充年龄和性别
                        input.setUser_gender(dimInfo.getString("GENDER"));
                        //计算年龄
                        String birthday = dimInfo.getString("BIRTHDAY");
                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
                        long age = (System.currentTimeMillis() - sdf.parse(birthday).getTime()) / (1000L * 60 * 60 * 24 * 365);
                        input.setUser_age(age);
                    }
                }, 60, TimeUnit.SECONDS);

        //userProvinceSkuWithUser.print(">>>>>>>");
        //关联地区维表
        SingleOutputStreamOperator<UserProvinceSku> userProvinceSkuWithProvDS = AsyncDataStream.unorderedWait(userProvinceSkuWithUser,
                new DimAsyncFunction<UserProvinceSku>("dim_base_province") {

                    @Override
                    public String getKey(UserProvinceSku input) {
                        return input.getProvince_id().toString();
                    }

                    @Override
                    public void join(UserProvinceSku input, JSONObject dimInfo) throws ParseException {
                        //补充省份相关信息
                        input.setProvince_name(dimInfo.getString("NAME"));
                        input.setProvince_area_code(dimInfo.getString("AREA_CODE"));
                        input.setProvince_iso_code(dimInfo.getString("ISO_CODE"));
                        input.setProvince_iso_3166_2(dimInfo.getString("iso_3166_2"));
                    }
                }, 60, TimeUnit.SECONDS);

        //关联SKU维表
        SingleOutputStreamOperator<UserProvinceSku> userProvinceSkuWithSkuDS = AsyncDataStream.unorderedWait(userProvinceSkuWithProvDS,
                new DimAsyncFunction<UserProvinceSku>("DIM_SKU_INFO") {
                    @Override
                    public String getKey(UserProvinceSku input) {
                        return input.getSku_id().toString();
                    }

                    @Override
                    public void join(UserProvinceSku input, JSONObject dimInfo) throws ParseException {
                        input.setSku_name(dimInfo.getString("SKU_NAME"));
                        input.setCategory3_id(dimInfo.getLong("CATEGORY3_ID"));
                        input.setSpu_id(dimInfo.getLong("SPU_ID"));
                        input.setTm_id(dimInfo.getLong("TM_ID"));
                    }
                }, 60, TimeUnit.SECONDS);
        //userProvinceSkuWithSkuDS.map(JSON::toJSONString).print(">>>>>>");

        //关联spu维表
        SingleOutputStreamOperator<UserProvinceSku> userProvinceSkuWithSpu = AsyncDataStream.unorderedWait(userProvinceSkuWithSkuDS,
                new DimAsyncFunction<UserProvinceSku>("DIM_SPU_INFO") {

                    @Override
                    public String getKey(UserProvinceSku input) {
                        return input.getSpu_id().toString();
                    }

                    @Override
                    public void join(UserProvinceSku input, JSONObject dimInfo) throws ParseException {
                        input.setSpu_name(dimInfo.getString("SPU_NAME"));
                    }
                }, 60, TimeUnit.SECONDS);

        //关联商品品类 3，2，1
        SingleOutputStreamOperator<UserProvinceSku> userProvinceSkuWithCate3 = AsyncDataStream.unorderedWait(userProvinceSkuWithSpu,
                new DimAsyncFunction<UserProvinceSku>("DIM_BASE_CATEGORY3") {
                    @Override
                    public String getKey(UserProvinceSku input) {
                        return input.getCategory3_id().toString();
                    }

                    @Override
                    public void join(UserProvinceSku input, JSONObject dimInfo) throws ParseException {
                        input.setCategory3_name(dimInfo.getString("NAME"));
                        input.setCategory2_id(dimInfo.getLong("CATEGORY2_ID"));
                    }
                }, 60, TimeUnit.SECONDS);

        SingleOutputStreamOperator<UserProvinceSku> userProvinceSkuWithCate2 = AsyncDataStream.unorderedWait(userProvinceSkuWithCate3,
                new DimAsyncFunction<UserProvinceSku>("DIM_BASE_CATEGORY2") {
                    @Override
                    public String getKey(UserProvinceSku input) {
                        return input.getCategory2_id().toString();
                    }

                    @Override
                    public void join(UserProvinceSku input, JSONObject dimInfo) throws ParseException {
                        input.setCategory2_name(dimInfo.getString("NAME"));
                        input.setCategory1_id(dimInfo.getLong("CATEGORY1_ID"));
                    }
                }, 60, TimeUnit.SECONDS);

        SingleOutputStreamOperator<UserProvinceSku> userProvinceSkuWithCate1 = AsyncDataStream.unorderedWait(userProvinceSkuWithCate2,
                new DimAsyncFunction<UserProvinceSku>("DIM_BASE_CATEGORY1") {
                    @Override
                    public String getKey(UserProvinceSku input) {
                        return input.getCategory1_id().toString();
                    }

                    @Override
                    public void join(UserProvinceSku input, JSONObject dimInfo) throws ParseException {
                        input.setCategory1_name(dimInfo.getString("NAME"));
                    }
                }, 60, TimeUnit.SECONDS);

        //关联品牌维度
        SingleOutputStreamOperator<UserProvinceSku> userProvinceSkuDimDS = AsyncDataStream.unorderedWait(userProvinceSkuWithCate1,
                new DimAsyncFunction<UserProvinceSku>("DIM_BASE_TRADEMARK") {
                    @Override
                    public String getKey(UserProvinceSku input) {
                        return input.getTm_id().toString();
                    }

                    @Override
                    public void join(UserProvinceSku input, JSONObject dimInfo) throws ParseException {
                        input.setTm_name(dimInfo.getString("TM_NAME"));
                    }
                }, 60, TimeUnit.SECONDS);
        userProvinceSkuDimDS.map(JSON::toJSONString).print(">>>>>");

        //TODO 8.将数据写出
        userProvinceSkuDimDS.addSink(ClickHouseUtil.getClickHouseSink("insert into dws_user_province_sku_10s values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"));

        //TODO 执行任务
        env.execute("UserProvinceSku10sApp");
    }
}
